package com.atguigu.day03;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

public class Flink04_Source_Custom {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //TODO 2.通过自定义数据源获取数据
        DataStreamSource<WaterSensor> dataStreamSource = env.addSource(new MySource()).setParallelism(2);

        dataStreamSource.print();

        env.execute();
    }
//    public static class MySource implements SourceFunction<WaterSensor>{
    //设置多并行度，需要实现以下接口
    public static class MySource implements ParallelSourceFunction<WaterSensor> {
        private boolean isRunning = true;
        private Random random = new Random();

        /**
         * 在这个方法中生成数据并通过ctx发送出去
         * @param ctx 发送数据
         * @throws Exception
         */
        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            while (isRunning){
                ctx.collect(new WaterSensor("sensor_"+random.nextInt(1000),System.currentTimeMillis(),random.nextInt(100)));
            }
        }

        /**
         * 通常在run中有while循环，通用来控制循环的终止，并且这个方法是flink内部自己调用的
         */
        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
